package com.lvmama.rhino.message;

import com.lvmama.rhino.message.RhinoMessage;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.DefaultDecoder;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;

/**
 * Created by wuyunpeng on 2016/4/26.
 */
public class RhinoMessageConsumer {

    private static final Logger LOGGER = Logger.getLogger(RhinoMessageConsumer.class);

    private static ConsumerConnector  consumer ;

    private ExecutorService executor;

    private static void init(Properties props){
        ConsumerConfig config = new ConsumerConfig(props);
        consumer = Consumer.createJavaConsumerConnector(config);
    }

    public RhinoMessageConsumer(Properties props,String group){
        props.put("group.id", group);
        init(props);
    }

    public RhinoMessageConsumer(Properties props){
        init(props);
    }

    /**
     * 消费对象类型消息
     * @param topic kafka主题
     * @param threadNum 消费线程数
     * @return
     */
    @SuppressWarnings("unchecked")
    public  void receive(String topic, Integer threadNum, final RhinoMessageProcesser processer){
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, threadNum);
        Map<String, List<KafkaStream<byte[], RhinoMessage>>> consumerMap = consumer.createMessageStreams(topicCountMap,new DefaultDecoder(null),new ObjectEncoderAndDecoder(null));
        List<KafkaStream<byte[], RhinoMessage>> streams = consumerMap.get(topic);
		for (final KafkaStream<byte[], RhinoMessage> kafkaStream : streams) {
			new Thread(new Runnable() {
				public void run() {
					for(MessageAndMetadata<byte[], RhinoMessage> mm : kafkaStream){
						RhinoMessage rhinoMessage = mm.message();
						//得到对象类型消息,做逻辑处理
                        processer.processer(rhinoMessage);
					}
				}
			}).start();
		}
    }

    /**
     * 关闭消费者
     */
    public void shutdown() {
        if (consumer != null) consumer.shutdown();
    }
}
